This plus the tech debt cleanup allows for the subscription handling to be greatly simplified, here and below in #assign

This was only ever used to encode the subscription info, which is now all handled by getTaskLags

This method could actually become private, except for a single test. I'm wondering if we can port that test to use getTaskLags instead. Aside from letting us make this private, that would probably improve our testing coverage, since I suppose that test was intended to unit test this class, meaning it should be testing this class's public API, not internal methods.

This method is getting refactored somewhat heavily in the next PR, where we actually collect the offsets and will lay in some heavy testing. Still trying to strike a balance between not writing/fixing up any tests until the end, and spending time on 50 tests per PR that are all rendered useless as soon as I start the next one.
But, I will tighten up this test

req: I understand why you called this a pair. However, it seems odd that this pair consists of three fields. Could you call it TaskLagTriple?

Ack, will fix this if it turns out nested structs aren't possible

Q: Is it not possible to use the struct TaskId here? If not, the versions field of TaskId (and of all nested fields?) should be set to 1-6, shouldn't they? TaskId is only used for the fields that are removed in version 7.

Working on finding someone who knows the code generation code to clear things up here (whether we can have nested structs, and what is the correct version for structs/inner fields)

Alright so the good/bad news is that this only works if the nested struct is an array (ie can do type: []TaskId but not type: TaskId). There's no reason for this to be the case, besides "didn't have the time and need to implement it" so we could in theory just add this ability.

I haven't looked into the code really so I'm not sure how much time that might take, so, I'm deciding to just encode as basic types for now. It's only two fields after all

Just to be clear, does this mean we're certain that for non-future members (current or older-versioned ones), the encoded "prevTasks" actually contains all the previous tasks?

I gather this is true from the SubscriptionInfo protcol:

    {
      "name": "prevTasks",
      "versions": "1-6",
      "type": "[]TaskId"
    }

But then, I'm a little mystified by the prior comment... why would "active tasks" not have been encoded with the cooperative protocol?

Let me give a quick history lesson to clarify:
By adding the ownedPartitions field to the subscription we were able to remove the encoded prevTasks and avoid duplicating info we could get from the new field for COOPERATIVE members. Note the ownedPartitions are the source of truth and may differ from the prevTasks that would have been encoded in edge cases (eg topic deletion, partitions lost)
In hindsight, aka a few weeks back when I started looking at the assignor code again, I realized this was just unnecessary and likely to lead to more trouble than it solves. So, now we just encode all tasks in the offset map regardless of rebalance protocol.
Ok, while explaining that I now realize we obviously still need to fill in the prevTasks from the ownedPartitions for members on 2.5/2.6 -- thanks Socrates :P

Either that, or my preference would actually be to inline both sentinel lags (with a comment explaining why that choice of sentinels).

To be honest, this also bugs me :)

What if we move org.apache.kafka.streams.tests.StreamsUpgradeTest.FutureStreamsPartitionAssignor into package org.apache.kafka.streams.processor.internals? Then package-private would continue to work fine.

My personal bias is that anytime you see // visible for testing, you're looking at a potential bug, because nothing prevents that comment from becoming false, and in fact, I have found such comments in our code base that were already false. Either this method is part of the public contract of the class, or it's not.

That said, if you really prefer it this way, we can keep it (although, I might ask you to review a clean-up PR later ;) )

This method could actually become private, except for a single test. I'm wondering if we can port that test to use getTaskLags instead. Aside from letting us make this private, that would probably improve our testing coverage, since I suppose that test was intended to unit test this class, meaning it should be testing this class's public API, not internal methods.

Oh, actually, here's the reason a constant sentinel is nice, but we didn't actually use it!

🤦‍♀

Since we invoke this method from a number of places, should we add a flag and make sure it only sets the state once?

bump

I realized the other callers actually don't need to call this at all, so now these only get called from the constructor

I'm probably missing the point here, but I think the idea of this class is that it should not change in response to changes in SubscriptionInfo. I think it's supposed to be a stand-in for the behavior of older Streams versions when the cluster has old and new members running at the same time. Maybe it doesn't really work that way, though, in which case, I might doubt the utility of this class at all, and instead recommend relying on the system tests. Can you comment?

Yes, it should... I asked a question about this on that class itself. It seems like you shouldn't have had to modify it at all.

I'm not sure this should be necessary either. IIUC, the "future" subscription info isn't supposed to really be a descendant of the current protocol, just a stand-in for some protocol version bigger than ours, in which case all that really matters is the version number. Its role is just to join the cluster and get downgraded to the "latest" version, in which case it should be able to defer to SubscriptionInfo.

Good point

bump

Do we want to try for a slightly more efficient encoding here, of Map[topicGroupId -> Map[partition -> offsetSum]], or do you think this is fine for now?

I was vaguely hoping that we'd add the ability to use nested structs in the near future and could move to use the TaskId struct here, but I suppose we may as well take advantage of this to go for the more efficient encoding when possible. Will do

looks like maybe we have duelling code formatters. Not sure which choice makes more sense.

Sounds like you think the other choice makes more sense 😜 Honestly neither of them looks that good to me but I'll revert the reformatting

I'm just a tiny bit uncomfortable with re-using that sentinel, because the correctness of our logic depends on the active sentinel being less than the standby sentinel, so it must be less than zero. Do we have a reason to believe that Task.LATEST_OFFSET would never change to a number that would spoil us here, such as zero?

I actually changed this based on working on the next PR, as Task#changelogOffsets uses this sentinel for exactly the same thing, ie an indicator that the task is running (and active). This is only used in computing the lag info for KIP-535, which has a similar desire to differentiate between a running task that is completely caught up and any other. So, I can't imagine this being changed -- but I can add a comment to the constant explaining it should always be negative (not sure why it's "-2" specifically, as opposed to "-1", do you?)